package com.alibaba.datax.plugin.reader.aliyunslsreader;

import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordSender;
import com.alibaba.datax.common.spi.Reader;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.plugin.reader.aliyunslsreader.util.LogHubReaderUtil;
import com.aliyun.openservices.loghub.client.ClientWorker;
import com.aliyun.openservices.loghub.client.config.LogHubConfig;
import com.aliyun.openservices.loghub.client.exceptions.LogHubClientWorkerException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * Created by dancoder on 2022.01.08
 */
public class AliyunSlsReader extends Reader {

    public static class Job extends Reader.Job {
        private static final Logger logger = LoggerFactory.getLogger(AliyunSlsReader.Job.class);

        private Configuration readerOriginConfig = null;

        private String endpoint;
        private String accessId;
        private String accessKey;
        private String project;
        private String logstore;
        private String consumeGroup;
        private Integer threadDuration;

        public Job() {
        }

        @Override
        public void init() {
            logger.debug("init begin...");

            this.readerOriginConfig = this.getPluginJobConf();
            this.validate();

            // init logHubConfig
            try {
                LogHubReaderUtil.initLogHubConfig(this.endpoint, this.accessId, this.accessKey,
                                this.project, this.logstore, this.consumeGroup);
            } catch (Exception e) {
                throw DataXException.asDataXException(AliyunSlsReaderErrorCode.RUNTIME_EXCEPTION, "阿里云SLS服务连接异常，请检查参数是否正确");
            }

            logger.debug("init ok end...");
        }

        private void validate() {
            this.endpoint = this.readerOriginConfig.getNecessaryValue(Key.ENDPOINT, AliyunSlsReaderErrorCode.CONFIG_INVALID_EXCEPTION);
            this.accessId = this.readerOriginConfig.getNecessaryValue(Key.ACCESSID, AliyunSlsReaderErrorCode.CONFIG_INVALID_EXCEPTION);
            this.accessKey = this.readerOriginConfig.getNecessaryValue(Key.ACCESSKEY, AliyunSlsReaderErrorCode.CONFIG_INVALID_EXCEPTION);
            this.project = this.readerOriginConfig.getNecessaryValue(Key.PROJECT, AliyunSlsReaderErrorCode.CONFIG_INVALID_EXCEPTION);
            this.logstore = this.readerOriginConfig.getNecessaryValue(Key.LOGSTORE, AliyunSlsReaderErrorCode.CONFIG_INVALID_EXCEPTION);
            this.consumeGroup = this.readerOriginConfig.getNecessaryValue(Key.CONSUME_GROUP, AliyunSlsReaderErrorCode.CONFIG_INVALID_EXCEPTION);
            this.threadDuration = Integer.parseInt(this.readerOriginConfig.getNecessaryValue(Key.THREAD_DURATION, AliyunSlsReaderErrorCode.CONFIG_INVALID_EXCEPTION));
        }

        @Override
        public void destroy() {
            logger.debug("destroy...");
        }

        @Override
        public List<Configuration> split(int adviceNumber) {
            logger.info("split begin...");
            List<Configuration> configurationList = new ArrayList<Configuration>();
            configurationList.add(this.readerOriginConfig.clone());
            logger.info("split end...");
            return configurationList;
        }
    }

    public static class Task extends Reader.Task {
        private static final Logger LOG = LoggerFactory.getLogger(Reader.Task.class);

        private String endpoint;
        private String accessId;
        private String accessKey;
        private String project;
        private String logstore;
        private String consumeGroup;
        private Integer threadDuration;

        private Configuration taskConfig;
        private LogHubConfig logHubConfig;

        @Override
        public void init() {
            this.taskConfig = this.getPluginJobConf();

            this.validate();
            // init logHubConfig
            try {
                this.logHubConfig = LogHubReaderUtil.initLogHubConfig(this.endpoint, this.accessId, this.accessKey,
                        this.project, this.logstore, this.consumeGroup);
            } catch (Exception e) {
                throw DataXException.asDataXException(AliyunSlsReaderErrorCode.RUNTIME_EXCEPTION, "阿里云SLS服务连接异常，请检查参数是否正确");
            }
        }

        @Override
        public void startRead(RecordSender recordSender) {
            ClientWorker worker = null;
            try {
                // 1. 初始化logHubProcessorFactory 并插入RecordSender，使消费线程内部可以往Datax channel中填充数据
                LogHubProcessorFactory logHubProcessorFactory = new LogHubProcessorFactory();
                logHubProcessorFactory.setRecordSender(recordSender);
                logHubProcessorFactory.setConfiguration(this.taskConfig);
                logHubProcessorFactory.setTaskPluginCollector(this.getTaskPluginCollector());

                // 2. 初始化clientWorker
                worker = new ClientWorker(logHubProcessorFactory, logHubConfig);

                // 3. 开始消费数据
                Thread thread = new Thread(worker);
                thread.start();

                // 4.
                // 持续一段时间（时长为配置的threadDuration），防止线程直接shutdown；
                // 不要设置太长时间，如果太长，该reader进程会一直执行到设置的时长直至结束。如果有多个任务同时进行，则会生成多个进程，对runner服务器会有压力
                // 假如外部采集配置是15m，此处间隔可为 1-5m
                // duration可控制在采集cron间隔的1/5内 ，并且需留出 5 阶段的时间
                TimeUnit.SECONDS.sleep(this.threadDuration);
                worker.shutdown();

                // 5. worker运行时会生成多个异步任务，shutdown后并不会直接kill掉，需要等待未执行完的任务安全退出，官方建议30s
                TimeUnit.SECONDS.sleep(30);
            } catch (LogHubClientWorkerException e) {
                worker.shutdown();
                throw DataXException.asDataXException(AliyunSlsReaderErrorCode.RUNTIME_EXCEPTION,
                        String.format("创建worker失败, 异常信息：%s", e.getMessage()));
            } catch (InterruptedException e) {
                worker.shutdown();
                throw DataXException.asDataXException(AliyunSlsReaderErrorCode.RUNTIME_EXCEPTION,
                        String.format("任务失败,异常信息： %s", e.getMessage()));
            }
        }

        private void validate() {
            this.endpoint = this.taskConfig.getNecessaryValue(Key.ENDPOINT, AliyunSlsReaderErrorCode.CONFIG_INVALID_EXCEPTION);
            this.accessId = this.taskConfig.getNecessaryValue(Key.ACCESSID, AliyunSlsReaderErrorCode.CONFIG_INVALID_EXCEPTION);
            this.accessKey = this.taskConfig.getNecessaryValue(Key.ACCESSKEY, AliyunSlsReaderErrorCode.CONFIG_INVALID_EXCEPTION);
            this.project = this.taskConfig.getNecessaryValue(Key.PROJECT, AliyunSlsReaderErrorCode.CONFIG_INVALID_EXCEPTION);
            this.logstore = this.taskConfig.getNecessaryValue(Key.LOGSTORE, AliyunSlsReaderErrorCode.CONFIG_INVALID_EXCEPTION);
            this.consumeGroup = this.taskConfig.getNecessaryValue(Key.CONSUME_GROUP, AliyunSlsReaderErrorCode.CONFIG_INVALID_EXCEPTION);
            this.threadDuration = Integer.parseInt(this.taskConfig.getNecessaryValue(Key.THREAD_DURATION, AliyunSlsReaderErrorCode.CONFIG_INVALID_EXCEPTION));
        }

        @Override
        public void destroy() {

        }
    }
}
